package simpleim

import (
	"context"
	"encoding/json"
	"github.com/IBM/sarama"
	"github.com/gorilla/websocket"
	"log"
	"net/http"
	"strconv"
	"time"
)

/*
WsGateway 服务网关节点
主要承担以下职责
1. 接收端请求
2. 转发请求
3. 注册消费消息
*/
type WsGateway struct {
	svc        *IMService
	InstanceId string
	client     sarama.Client
	store      *ConnStore
}

// Start 启动网关服务
func (ws *WsGateway) Start(port string) {
	// websocket协议升级
	mux := http.NewServeMux()
	upgrader := &websocket.Upgrader{
		// 允许跨域请求
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}
	mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			// 升级失败，直接退出
			log.Printf("err:%v", err)
			w.Write([]byte("初始化Websocket失败！"))
			return
		}
		// 记录一下关联关系
		uid := ws.Uid(r)
		log.Printf("uid:%v", uid)
		ws.store.Store(uid, conn)
		// 开启协程监听数据
		go func() {
			defer func() {
				ws.store.Delete(uid)
			}()
			for {
				typ, msgBytes, err := conn.ReadMessage()
				switch err {
				case nil:
				case context.DeadlineExceeded:
					// 超时重新读
					continue
				default:
					log.Println("读取消息发生错误: ", err)
					return
				}

				switch typ {
				case websocket.TextMessage, websocket.BinaryMessage:
					// 转发消息
					var msg Message
					err = json.Unmarshal(msgBytes, &msg)
					if err != nil {
						log.Printf("数据格式解析异常: %v", err)
						continue
					}
					ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
					err = ws.svc.handle(ctx, uid, msg)
					cancel()
					if err != nil {
						// 告诉端哪条消息发送失败
						data, err := json.Marshal(Message{
							Seq:     msg.Seq,
							Type:    "result",
							Content: "failed",
						})
						if err != nil {
							log.Printf("序列化消息失败: %v, seq: %v", err, msg.Seq)
						}
						err = conn.WriteMessage(websocket.TextMessage, data)
						if err != nil {
							// 消息发送失败
							log.Printf("消息发送失败: %v, msg: %v", err, msg)
						}
					}
				case websocket.CloseMessage:
					conn.Close()
				default:
					log.Fatalf("非法数据格式: %v", typ)
				}
			}
		}()
	})
	ws.SubscribeMsg()
	http.ListenAndServe(":"+port, mux)
}

// SubscribeMsg 每个网关节点都需要订阅消息
func (ws *WsGateway) SubscribeMsg() error {
	// 每个节点都是单独的消费者组
	cg, err := sarama.NewConsumerGroupFromClient(ws.InstanceId, ws.client)
	if err != nil {
		log.Printf("创建消费者组失败, instanceId:%v", ws.InstanceId)
		return err
	}
	go func() {
		er := cg.Consume(context.Background(), []string{EventName}, &ConsumerHandler{store: ws.store})
		if er != nil {
			log.Printf("退出消息监听循环:%v", er)
		}
	}()
	return nil
}

// Uid 获取当前登录用户的uid（一般从token或者session中拿）
// 这里模拟用header拿
func (ws *WsGateway) Uid(httpRequest *http.Request) int64 {
	uidStr := httpRequest.Header.Get("Uid")
	uid, _ := strconv.ParseInt(uidStr, 10, 64)
	return uid
}

// Message 消息结构定义
type Message struct {
	Seq     string `json:"seq"`     // 消息序列号
	ID      int64  `json:"id"`      // 消息id（后端的id，用于支持消息引用等功能）
	Cid     string `json:"cid"`     // 群组id
	Type    string `json:"type"`    // 消息类型（比如video、text）
	Content string `json:"content"` // 消息体
}
